package com.pixelengine;
// 使用js进行瓦片计算，结果写入HBase
// 2022-3-27 created


import com.google.gson.Gson;
import com.pixelengine.DataModel.*;
import com.pixelengine.tools.JRoi2Loader;
import com.pixelengine.tools.JScriptTools;
import com.pixelengine.tools.JTileRangeTool;
import com.pixelengine.tools.Roi2HsegTlv2LonLatExtent;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.broadcast.Broadcast;
import scala.Serializable;

import java.io.File;
import java.io.FileOutputStream;
import java.io.OutputStream;
import java.time.LocalDateTime;
import java.util.*;

public class WTileComputing2HBaseProcessor implements Serializable {


    public String task17configFile ;
    public String orderJsonFile ;
    public String resultJsonFile ;

    protected boolean writeResultJson(int state, int mypid, String msg)
    {
        if (resultJsonFile.isEmpty()) {
            System.out.println("resultJsonFile is empty.");
            return false ;
        } else {
            try {
                OutputStream outputStream = new FileOutputStream(resultJsonFile);
                String content = "{\"state\":" + String.valueOf(state)
                        + ",\"mypid\":" + String.valueOf(mypid)
                        + ",\"message\":\"" + msg + "\"}";
                outputStream.write(content.getBytes());
                outputStream.close();
                return true ;
            } catch (Exception ex) {
                System.out.println("WTileComputing2HBaseProcessor.writeResultJson exception:" + ex.getMessage());
                return false ;
            }
        }
    }

    public int runit()
    {
        //load resource from config.json in jar
        System.out.println("WTileComputing2HBaseProcessor running. parsing task17config.json");
        String task17configJsonText = WTextFile.readFileAsString(task17configFile) ;
        WConfig wconfig = null ;
        {
            WConfig.initWithString(task17configJsonText);
            wconfig = WConfig.getSharedInstance() ;
        }
        System.out.println("**************** Check zookeeper, spark, mysql info**************");
        System.out.println("zookeeper:" + wconfig.zookeeper);
        System.out.println("sparkmaster:"+wconfig.sparkmaster);
        System.out.println("connstr:" + wconfig.connstr);
        System.out.println("user:" + wconfig.user);
        System.out.println("pwd:" + wconfig.pwd);
        System.out.println("**************** *************************** *******************");
        String sparkmaster = wconfig.sparkmaster ;
        String zookeeper = wconfig.zookeeper ;



        //init mysql
        JRDBHelperForWebservice.init(wconfig.connstr,wconfig.user,wconfig.pwd);
        JRDBHelperForWebservice rdb = new JRDBHelperForWebservice() ;

        //init spark+hbase
        String appname = new File(orderJsonFile).getName() ;
        System.out.println("use filename as appname:" + appname);
        //Configuration conf = HBaseConfiguration.create();
        //conf.set("hbase.zookeeper.quorum", zookeeper );//must has this code for hbase.
        SparkConf sparkconf = new SparkConf().
                setAppName(appname).setMaster(sparkmaster);
        JavaSparkContext jsc = new JavaSparkContext(sparkconf);
        //JavaHBaseContext hbaseContext = new JavaHBaseContext(jsc, conf);

        //show pixelengine core version
        HBasePeHelperCppConnector cc = new HBasePeHelperCppConnector();
        System.out.println("pe core version: " + cc.GetVersion() );

        //load input order.json
        System.out.println("load task order jsonfile "+orderJsonFile);
        Gson gson = new Gson() ;
        JTileComputing2HBaseOrder tcHbOrder = gson.fromJson(
                WTextFile.readFileAsString(orderJsonFile),JTileComputing2HBaseOrder.class) ;
        //为了保留之前的逻辑，首先检查hpid和mpid是否为0，如果为0，那么从mpid_hpid 获取这两个值，并认为这两个值一致。
        if( tcHbOrder.mpid==0 ) {
            tcHbOrder.mpid = tcHbOrder.mpid_hpid ;
        }
        if( tcHbOrder.hpid == 0 ) {
            tcHbOrder.hpid = tcHbOrder.mpid_hpid ;
        }
        if( tcHbOrder.out_mydt0 == 0 ) {
            tcHbOrder.out_mydt0 = tcHbOrder.out_hcol;
        }
        if( tcHbOrder.out_mydt1 == 0 ) {
            //如果没有设置out_mydt1,默认out_mydt0加一天
            int ymd0 = (int)(tcHbOrder.out_hcol/1000000L);
            int year0 = ymd0/10000;
            int month0 = ymd0/100%100 ;
            int day0 = ymd0%100;
            long hms0 = (tcHbOrder.out_hcol%1000000L) ;
            //2024-1-22 bugfixed for v2.6.1.0
            if( year0==0 || month0==0 || day0==0 ){
                tcHbOrder.out_mydt1 = tcHbOrder.out_mydt0+1;
            }else{
                Date tdt0 = new GregorianCalendar(year0,month0-1,day0).getTime() ;
                LocalDateTime tdt1 = LocalDateTime.from(tdt0.toInstant()).plusDays(1);
                long dt1value = tdt1.getYear()*10000+tdt1.getMonthValue()*100+tdt1.getDayOfMonth() ;
                dt1value=dt1value*1000000L + hms0 ;
                tcHbOrder.out_mydt1 = dt1value;
            }
        }

        //计算0，0，0号瓦片，获取 dsnameArr 列表，与时间列表 dtArr ，输出类型 outdatatype
        String scriptText = WTextFile.readFileAsString(tcHbOrder.jsfile) ;
        WComputeZeroTile computerForZeroTile = new WComputeZeroTile() ;
        TileComputeResultWithRunAfterInfo tileResultWithRunAfterInfo =
                computerForZeroTile.computeZeroTile(
                        scriptText,
                        tcHbOrder.dt,
                        tcHbOrder.sdui
                ) ;
        if( computerForZeroTile==null ){
            writeResultJson(20,0,"bad computerForZeroTile result.");
            return 20 ;
        }

        //output info for zero tile computing
        ArrayList<JDsnameDts> dsnameDtsArr = tileResultWithRunAfterInfo.getDsnameDtsArray() ;
        if( dsnameDtsArr.size()==0){
            writeResultJson(21,0,"the script should use one dataset at least.");
            return 21 ;
        }
        ArrayList<String> roi2Arr = tileResultWithRunAfterInfo.getRoi2Array() ;
        System.out.println("dsname0,dt0:" + dsnameDtsArr.get(0).dsname+","+dsnameDtsArr.get(0).dtarr.get(0));
        if( roi2Arr.size()>0) System.out.println("roi2-0:"+roi2Arr.get(0)) ;
        System.out.println("outDatatype:"+tileResultWithRunAfterInfo.dataType);
        System.out.println("outNumband:" + tileResultWithRunAfterInfo.nbands);

        //dsnameArr proj属性必须一致，否则瓦片对不上,通过dsname计算 最小值maxlevel
        String projStr = "" ;
        Long hcolForDs0Dt0 =  dsnameDtsArr.get(0).dtarr.get(0) ;
        int maxZoom = 12;//这里以后要修改成根据输入决定最大层级，对于小于该层级的产品要下采样计算2023-11-29
        int firstGoodStyleid = 0 ;
        JProduct[] productInfoArray = new JProduct[dsnameDtsArr.size()] ;
        for( int ids = 0 ; ids< dsnameDtsArr.size() ; ++ ids ){
            JProduct pinfo = rdb.rdbGetProductInfoByName(dsnameDtsArr.get(ids).dsname) ;
            productInfoArray[ids] = pinfo ;
            if( ids==0 ){
                projStr = pinfo.proj ;
                maxZoom = pinfo.maxZoom ;
            }else
            {
                if( projStr.equals( pinfo.proj) ==false ){
                    writeResultJson(22, 0,"bad proj0 "+projStr + " and proj1 "+pinfo.proj);
                    return 22;
                }
                if( pinfo.maxZoom < maxZoom) maxZoom = pinfo.maxZoom ;//取值最小的level
            }
            if( pinfo.styleid>=0 &&  firstGoodStyleid == 0 ){
                firstGoodStyleid = pinfo.styleid;
            }
        }

        //如果没有roi，那么需要确定哪些瓦片参与计算？还是说在瓦片计算中自己判断
        //如果需要在瓦片计算中自己判断，就需要加速瓦片筛选的效率，一秒1个肯定不行
        //以z=12为例，总计4096x2048=8million瓦片，一秒1个也要2330小时
        //我需要确认一下Hbase获取全部key的效率 2023-11-29
        // 2023-11-30 经过实验HBase更适合随机读取，不适合范围读取，所以Get平均效率要高一些，
        // 而且气候中心服务器z=8的时候遍历瓦片约1s，超过z=10的时候需要时间较长不适合应用

        //通过数据集和时间计算经纬度范围 extent，取最大范围
        //对于大于等于8的层级通过z=8计算范围，反之计算当前z的范围
        double outExtentLeft = -180 ;
        double outExtentRight = 180 ;
        double outExtentBottom = -90 ;
        double outExtentTop = 90 ;
        boolean zExtentInited = false ;
        int zForExtent = Math.min(8, maxZoom) ;
        JHBaseTileExtent jhbaseTileExtent = new JHBaseTileExtent( zookeeper ) ;
        for( int ids = 0 ; ids< dsnameDtsArr.size() ; ++ ids){
            JProduct pinfo = productInfoArray[ids] ;
            JDsnameDts dsdts = dsnameDtsArr.get(ids) ;

            for( long currdt: dsdts.dtarr ) {

                JTileXyzLonLat[] tileLLs = jhbaseTileExtent.computeTileExtent(pinfo.hbaseTable.hTableName,"tiles",
                        pinfo.bandList.get(0).hPid,
                        pinfo.hbaseTable.hPidByteNum,
                        pinfo.hbaseTable.hYXByteNum,
                        currdt,
                        zForExtent) ;

                if( zExtentInited==false ) {
                    zExtentInited = true ;
                    outExtentLeft = tileLLs[0].leftTopLon ;
                    outExtentTop =  tileLLs[0].leftTopLat ;
                    outExtentRight = tileLLs[1].rightBottomLon;
                    outExtentBottom = tileLLs[1].rightBottomLat ;
                }else{
                    //max the extent
                    outExtentLeft = Math.min(tileLLs[0].leftTopLon,outExtentLeft) ;
                    outExtentTop =  Math.max(tileLLs[0].leftTopLat,outExtentTop) ;
                    outExtentRight = Math.max(tileLLs[1].rightBottomLon,outExtentRight);
                    outExtentBottom = Math.min(tileLLs[1].rightBottomLat,outExtentBottom) ;
                }
            }
        }
        System.out.println("dsdts tile lonlat extent:" + outExtentLeft+","+outExtentRight+","+outExtentTop+","+outExtentBottom);

        //通过roi计算经纬度范围 extent，与前面范围取交集
        boolean useOrderRoi = false ;
        byte[] orderRoi2TlvData = new byte[1] ;//a nearly empty byte array for broadcast if no order roi
        if(tcHbOrder.roi.length()>0) {
            System.out.println("calculate extent of roi in order.");
            roi2Arr.add( tcHbOrder.roi) ;
            useOrderRoi = true ;
            orderRoi2TlvData = JRoi2Loader.loadData(tcHbOrder.roi) ;
            if( orderRoi2TlvData==null ){
                writeResultJson(23,0,"bad order roi data: user add a roi in order but we load a null data, the roi input is "+tcHbOrder.roi);
                return 23 ;
            }
        }
        if( roi2Arr.size()> 0 ){
            System.out.println("calculate extent of roi in codes.");
            Roi2HsegTlv2LonLatExtent.Extent tileExtent = Roi2HsegTlv2LonLatExtent.expandHalfPixel(
                    Roi2HsegTlv2LonLatExtent.computeExtent(
                            roi2Arr.toArray(new String[0])
                    )
            ) ;
            outExtentLeft = tileExtent.left ;
            outExtentRight = tileExtent.right ;
            outExtentTop = tileExtent.top ;
            outExtentBottom = tileExtent.bottom ;
        }
        System.out.println("use output lonlat extent:" + outExtentLeft+","+outExtentRight+","+outExtentTop+","+outExtentBottom);


        //通过 extent 和 0-maxlevel  计算每个level的 tile_extentArray
        //每个JTileArray给 JTileArray.DEFAULTSIZE 个瓦片计算
        int totalTileCount = 0 ;
        ArrayList<JTileArray> tileXyzArrayGroup = new ArrayList<>(128) ;
        JTileArray cursorTileArray = null ;
        for(int iz = 0 ; iz <= maxZoom ; ++ iz )
        {
            JTileRangeTool.TileXYRange range1 = JTileRangeTool.computeTileRangeByLonglatExtent(
                    outExtentLeft,outExtentRight,
                    outExtentTop,outExtentBottom,
                    iz,256) ;
            for(int ity = range1.ymin ; ity < range1.ymax+1; ++ ity ){
                for(int itx = range1.xmin ; itx < range1.xmax+1; ++ itx ){
                    if( cursorTileArray == null ) {
                        cursorTileArray = new JTileArray() ;
                    }
                    totalTileCount++;
                    cursorTileArray.tileArray.add(new TileXYZ(iz,ity,itx)) ;
                    if( cursorTileArray.tileArray.size() == JTileArray.DEFAULTSIZE ) {
                        tileXyzArrayGroup.add( cursorTileArray ) ;
                        cursorTileArray = null ;
                    }
                }
            }
        }
        if( cursorTileArray!=null ) {
            tileXyzArrayGroup.add( cursorTileArray ) ;
            cursorTileArray = null ;
        }
        System.out.println("output tile count will be :"+ totalTileCount );
        System.out.println("use TileArray Group will be :"+ tileXyzArrayGroup.size() );

        //shared data
        String scriptWithSDUI = JScriptTools.assembleScriptWithSDUI(scriptText,tcHbOrder.sdui) ;
        Broadcast<String> broadcastScriptWithSDUI = jsc.sc().broadcast(
                scriptWithSDUI, scala.reflect.ClassManifestFactory.fromClass(String.class)
        ) ;


        String extraText = "{\"datetime\":"+String.valueOf(tcHbOrder.dt)+"}" ;
        Broadcast<String> broadcastExtraText = jsc.sc().broadcast(
                extraText ,
                scala.reflect.ClassManifestFactory.fromClass(String.class)) ;
        Broadcast<Integer> broadcastHpid = jsc.sc().broadcast(
                tcHbOrder.hpid ,
                scala.reflect.ClassManifestFactory.fromClass(Integer.class)) ;
        Broadcast<Integer> broadcastHpidLen = jsc.sc().broadcast(
                tcHbOrder.out_hpidlen ,
                scala.reflect.ClassManifestFactory.fromClass(Integer.class)) ;
        Broadcast<Integer> broadcastYxLen = jsc.sc().broadcast(
                tcHbOrder.out_xylen ,
                scala.reflect.ClassManifestFactory.fromClass(Integer.class)) ;
        Broadcast<Long> broadcastHcol = jsc.sc().broadcast(
                tcHbOrder.out_hcol ,
                scala.reflect.ClassManifestFactory.fromClass(Long.class)) ;
        Broadcast<String> broadcastHtable = jsc.sc().broadcast(
                tcHbOrder.out_htable ,
                scala.reflect.ClassManifestFactory.fromClass(String.class)) ;
        Broadcast<String> broadcastZk = jsc.sc().broadcast(
                wconfig.zookeeper ,
                scala.reflect.ClassManifestFactory.fromClass(String.class)) ;
        Broadcast<String> broadcastMyConnStr = jsc.sc().broadcast(
                wconfig.connstr ,
                scala.reflect.ClassManifestFactory.fromClass(String.class)) ;
        Broadcast<String> broadcastMyUser = jsc.sc().broadcast(
                wconfig.user ,
                scala.reflect.ClassManifestFactory.fromClass(String.class)) ;
        Broadcast<String> broadcastMyPwd = jsc.sc().broadcast(
                wconfig.pwd ,
                scala.reflect.ClassManifestFactory.fromClass(String.class)) ;
        Broadcast<String> broadcastTask17ConfigText = jsc.sc().broadcast(
                task17configJsonText ,
                scala.reflect.ClassManifestFactory.fromClass(String.class)) ;
        Broadcast<Boolean> broadcastUseOrderRoi = jsc.sc().broadcast(
                useOrderRoi ,
                scala.reflect.ClassManifestFactory.fromClass(Boolean.class)) ;
        Broadcast<byte[]> broadcastOrderRoiData = jsc.sc().broadcast(
                orderRoi2TlvData ,
                scala.reflect.ClassManifestFactory.fromClass(byte[].class)) ;
        Broadcast<Double> broadcastFillData = jsc.sc().broadcast(
                tcHbOrder.filldata ,
                scala.reflect.ClassManifestFactory.fromClass(Double.class)) ;

        //build RDD for tile computing
        System.out.println("task16_numslices: "+WConfig.getSharedInstance().task16_numslices);
        System.out.println("each slice process TileRDD: "+
                String.valueOf(tileXyzArrayGroup.size()/WConfig.getSharedInstance().task16_numslices) );
        JavaRDD<JTileArray> tileArrRdds = jsc.parallelize(tileXyzArrayGroup
                , WConfig.getSharedInstance().task16_numslices) ;
        //every tile do script computing, with a Integer return value

        //return 返回结果是成功的数量
        JavaRDD<Integer> tileResRdd = tileArrRdds.map(
                new Function<JTileArray, Integer>() {
                    @Override
                    public Integer call(JTileArray tileArr) throws Exception {
                        String brtask17configJsonText=broadcastTask17ConfigText.value() ;
                        WConfig.initWithString(brtask17configJsonText);
                        String myconn = broadcastMyConnStr.value();
                        String myuser = broadcastMyUser.value();
                        String mypwd = broadcastMyPwd.value() ;
                        Boolean brUseOrderRoi = broadcastUseOrderRoi.value() ;
                        byte[] brOrderRoiData = broadcastOrderRoiData.value() ;
                        double brFillData = broadcastFillData.value() ;

                        Integer brhpidblen = broadcastHpidLen.value() ;
                        Integer brhpid = broadcastHpid.value() ;
                        Integer bryxblen = broadcastYxLen.value() ;
                        long brhcol = broadcastHcol.value() ;

                        JRDBHelperForWebservice.init(myconn,myuser,mypwd);

                        String scirptWithSdui = broadcastScriptWithSDUI.value();
                        String extraText = broadcastExtraText.value() ;
                        HBasePeHelperCppConnector cc1 = new HBasePeHelperCppConnector();
                        Integer goodcnt = 0 ;
                        ArrayList<Put> putlist = new ArrayList<>(20) ;
                        byte[] famibytes = "tiles".getBytes() ;
                        byte[] hcolbytes = Bytes.toBytes(brhcol);
                        for( TileXYZ tilexyz : tileArr.tileArray) {
                            TileComputeResult tileResult1=
                                    cc1.RunScriptForTileWithoutRenderWithExtra(
                                            "com/pixelengine/HBasePixelEngineHelper",
                                            scirptWithSdui,
                                            extraText,
                                            tilexyz.z,tilexyz.y ,tilexyz.x
                                    ) ;
                            if(tileResult1!=null ){
                                //good
                                TileComputeResult newTCR = null ;
                                if( brUseOrderRoi==true ){
                                    newTCR = cc1.ClipTileComputeResultByHsegTlv(
                                            "com/pixelengine/HBasePixelEngineHelper",
                                            tileResult1 ,
                                            brOrderRoiData,
                                            brFillData
                                    ) ;
                                }else{
                                    newTCR = tileResult1 ;
                                }
                                byte[] outRowkey = WHBaseUtil.GenerateRowkey(
                                        brhpidblen ,
                                        brhpid ,
                                        bryxblen ,
                                        tilexyz.z,
                                        tilexyz.y,
                                        tilexyz.x) ;
                                Put put1 = new Put( outRowkey ) ;
                                put1.addColumn(
                                        famibytes ,
                                        hcolbytes ,
                                        newTCR.binaryData ) ;
                                putlist.add(put1) ;
                                goodcnt++;

                            }
                        }// end of for tileArr
                        if( putlist.size()>0 ) {
                            String brHtable = broadcastHtable.value() ;
                            String brzookeeper = broadcastZk.value() ;
                            Configuration conf = HBaseConfiguration.create();
                            conf.set("hbase.zookeeper.quorum",brzookeeper);
                            Connection hbaseConnection = ConnectionFactory.createConnection(conf) ;
                            Table outputTable = hbaseConnection.getTable( TableName.valueOf(brHtable) ) ;
                            outputTable.put( putlist) ;
                            outputTable.close();
                            hbaseConnection.close();
                        }
                        return goodcnt ;
                    }
                }) ;


        //trigger spark to run
        List < Integer > allTcStatus = tileResRdd.collect();
        ListIterator<Integer> tcStatusIter = allTcStatus.listIterator() ;
        int goodTcCount = 0 ;
        while(tcStatusIter.hasNext()){
            goodTcCount += tcStatusIter.next() ;
        }
        System.out.println("rdd finished, good tile computing count:"+String.valueOf(goodTcCount));
        System.out.println("rdd finished, bad tile computing count:"+String.valueOf( totalTileCount - goodTcCount));

        //add band records
        String someWarningInfo = "";
        boolean bandsok = rdb.writeProductBandRecord(
                tcHbOrder.mpid,
                tcHbOrder.hpid,
                tileResultWithRunAfterInfo.nbands,0,255,tcHbOrder.filldata) ;
        if( bandsok==false ){
            System.out.println("Warning: failed to write band infos into mysql. may be it already have bands info?");
            someWarningInfo+="Warning: failed to write band infos into mysql. may be it already have bands info?"
                    +" mpid "+tcHbOrder.mpid
                    +", hpid "+tcHbOrder.hpid
                    +", nbands "+tileResultWithRunAfterInfo.nbands
                    + ". " ;
            //no need exit 存在多次往一个mpid入库波段的情况，所以这里只是提示一下，不再退出程序
        }else{
            System.out.println("write bands records ok.");
        }
        //add dataitem record
        int dataitemId = rdb.writeProductDataItem(
                tcHbOrder.mpid,
                tcHbOrder.out_hcol,
                tcHbOrder.out_mydt0,//2023-11-28
                tcHbOrder.out_mydt1,//2023-11-28
                outExtentLeft,
                outExtentRight,outExtentTop,outExtentBottom) ;
        if( dataitemId<0 ){
            System.out.println("failed to write data item into mysql. maybe it already has a dateitem?");
            someWarningInfo+="Warning: failed to write data item into mysql. maybe it already has a dateitem?"
                    +" mpid "+tcHbOrder.mpid
                    +", hcol "+tcHbOrder.out_hcol
                    +", dt0 "+tcHbOrder.out_mydt0
                    +", dt1 "+tcHbOrder.out_mydt1
                    + ". " ;
        }else{
            System.out.println("write data item records ok:" + dataitemId);
        }

        //update mysql tbproduct record

        boolean updatepdtok = rdb.updateProductInfo(
                tcHbOrder.mpid,
                projStr,
                0,
                maxZoom,
                tileResultWithRunAfterInfo.dataType,
                0,
                tcHbOrder.out_htable,
                256,256,
                "deflate" ,
                firstGoodStyleid
        ) ;
        if( updatepdtok==false){
            System.out.println("failed to update pdt info.");
            writeResultJson(26,0,"failed to update pdt info.");
            return 26 ;
        }


        //done.
        someWarningInfo+=" good tiles:"+goodTcCount+", bad tiles:"+(totalTileCount-goodTcCount)+". " ;
        writeResultJson(0,tcHbOrder.mpid_hpid,someWarningInfo);
        System.out.println("*\n*\n*\n");
        System.out.println("mpid: " + tcHbOrder.mpid);
        System.out.println("hpid: " + tcHbOrder.hpid);
        System.out.println("hcol: " + tcHbOrder.out_hcol);
        System.out.println("dt0: " + tcHbOrder.out_mydt0);
        System.out.println("dt1: " + tcHbOrder.out_mydt1);
        System.out.println("good: "+goodTcCount);
        System.out.println("bad: "+(totalTileCount-goodTcCount));
        System.out.println("*\n*\n*\n");
        System.out.println("done");
        return 0 ;
    }
}
